package com.niit.service

import java.io.{FileWriter, PrintWriter}
import java.text.SimpleDateFormat

import com.niit.bean.AdClickData
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.dstream.DStream

import scala.collection.mutable.ListBuffer

class TimeCountService {

  def dataAnalysis(data:DStream[AdClickData]): Unit ={
    //1. 点击广告时间 进行转换
    /*
      13位的时间戳：1761348431273 == > 2025 05-24 10:43:12
      2025 05-24 10:43:12   ==> 2025 05-24 10:43:10
      2025 05-24 10:43:19   ==> 2025 05-24 10:43:10

      1761348431273 == > 2025 05-24 10:43:12  .splict(":")[2][1] =0  不建议这么操作

      1s 等于 1000毫秒  10秒 等于 10 * 1000
       1761348430000  == 2025 05-24 10:43:10

       1761348431273 / 10000 = 176134843 * 10000 = 1761348430000
     */

    //每10秒进行一次数据的统计和汇总
    val reduceDS = data.map(line=>{
      val ts: Long = line.ts.toLong  //1761348431273  59:53
      val newts = ts / 10000 * 10000  // 1761348430000  59:50
      (newts,1)  // 整秒的时间作为Key ,每一条数据就表示点击广告一次 值 1
    }).reduceByKeyAndWindow(
      (x:Int,y:Int)=>{
        x + y  //10秒内 的点击次数进行相加
      },
      Seconds(60),
      Seconds(10)
    )

    //2.对数据进行格式化，并发送的前端页面
    reduceDS.foreachRDD(rdd=>{
       //2.1新建列表，用来存储格式化的数据
      val list = ListBuffer[String]()
      //2.2将获取的数据 根据时间(Key)进行升序排序，返回是一个数组
      val datas: Array[(Long, Int)] = rdd.sortByKey(true).collect()
      //2.3得到数组，进行遍历。并根据模式匹配对数据进行格式化
      datas.foreach{
        case (time,count)=>{
          //将时间戳转成 HH:mm:ss 的格式
          val sdf = new SimpleDateFormat("HH:mm:ss")
          val timeStr = sdf.format(new java.util.Date(time.toLong))// 10:43:10
          //将显示到 Html页面
          //2.4组装数据  根据  { "xtime":"59:50", "yval":"88" } 进行数据的组装
          list.append(s"""{"xtime":"${timeStr}","yval":"${count}" } """)
        }
      }
      //将数据输出到 adclick.json
      val out = new PrintWriter( new FileWriter("data/adclick.json") )
      out.println("["+list.mkString(",")+"]")
      out.flush()
      out.close()
    })

  }

}
